package org.onion.pattern.event.springboot.rocketmq.consume;


import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetRouteInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;

import java.util.Map;
import java.util.concurrent.ConcurrentMap;


@Slf4j
@Setter
@Getter
class _Consumer extends DefaultMQPushConsumer {

    private boolean enabledAutoCreateTopic = true;
    private int     topicQueueNums;


    public _Consumer() {
        super();
    }

    public _Consumer(String consumerGroup) {
        this(null, consumerGroup);
    }

    public _Consumer(String namespace, String consumerGroup) {
        super(namespace, consumerGroup, null, new AllocateMessageQueueAveragely(), true, null);
    }

    public _Consumer(String consumerGroup, RPCHook rpcHook) {
        super(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(), true, null);
    }

    public _Consumer(String namespace, String consumerGroup, RPCHook rpcHook) {
        super(namespace, consumerGroup, rpcHook, new AllocateMessageQueueAveragely(), true, null);
    }

    @Override
    public void start() throws MQClientException {
        super.start();
        autoCreateTopic();
    }

    @Override
    public void subscribe(String topic, String subExpression) throws MQClientException {
//    mqClientAPIImpl.createTopic();
//    this.defaultMQPushConsumerImpl.getmQClientFactory().getMQClientAPIImpl().createTopic();
        super.subscribe(topic, subExpression);
    }

    @Override
    public void subscribe(String topic, String fullClassName, String filterClassSource)
            throws MQClientException {
        super.subscribe(topic, fullClassName, filterClassSource);
    }

    @Override
    public void subscribe(String topic, MessageSelector messageSelector) throws MQClientException {
        super.subscribe(topic, messageSelector);
    }

    protected void autoCreateTopic() {
        if (!enabledAutoCreateTopic) {
            return;
        }

        final MQClientAPIImpl mqClientAPIImpl = this.getDefaultMQPushConsumerImpl().getmQClientFactory()
                .getMQClientAPIImpl();
        final ConcurrentMap<String, SubscriptionData> subscriptionInner = this
                .getDefaultMQPushConsumerImpl()
                .getRebalanceImpl().getSubscriptionInner();
        for (String topicName : subscriptionInner.keySet()) {
            if (topicName.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                continue;
            }
            final RemotingClient remotingClient = mqClientAPIImpl.getRemotingClient();
            GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
            requestHeader.setTopic(topicName);
            RemotingCommand request = RemotingCommand
                    .createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);

            TopicConfig topicConfig = new TopicConfig();
            topicConfig.setWriteQueueNums(topicQueueNums);
            topicConfig.setReadQueueNums(topicQueueNums);
            topicConfig.setTopicName(topicName);

            try {
                RemotingCommand response = remotingClient.invokeSync(null, request, 3 * 1000);
                if (response.getCode() == ResponseCode.TOPIC_NOT_EXIST) {
                    RemotingCommand cluster_request = RemotingCommand
                            .createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);

                    RemotingCommand cluster_response = remotingClient
                            .invokeSync(null, cluster_request, 3 * 1000);
                    if (cluster_response.getCode() == ResponseCode.SUCCESS) {
                        ClusterInfo clusterInfo = ClusterInfo
                                .decode(cluster_response.getBody(), ClusterInfo.class);

                        final Map<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
                        for (BrokerData brokerData : brokerAddrTable.values()) {
                            mqClientAPIImpl
                                    .createTopic(brokerData.selectBrokerAddr(),
                                            RocketMQ.AUTO_CREATE_TOPIC_KEY_TOPIC, topicConfig,
                                            3 * 1000);
                            log.info("【事件】主题创建成功,cluster:{},broker:{},topic:{}"
                                    , brokerData.getCluster()
                                    , brokerData.getBrokerName()
                                    , topicName);
                        }

                    }
                }

            } catch (Exception e) {
                log.error("【事件】主题创建失败 {}", topicConfig, e);
            }
        }
    }

}
